package com.datastream.process;

import com.alibaba.fastjson.JSONArray;
import com.k2data.platform.ddm.sdk.builder.KMXRecordBuilder;
import com.k2data.platform.ddm.sdk.builder.TSDBRecordBuilder;
import com.k2data.platform.ddm.sdk.builder.TSDWRecordBuilder;
import com.k2data.platform.ddm.sdk.client.KMXClient;
import com.k2data.platform.ddm.sdk.exception.NullRecordException;
import com.k2data.platform.ddm.sdk.exception.SendException;
import configuration.KMXServerConfig;
import entity.Topic1Data;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.List;
import java.util.Set;

/**
 * Created by kzdatd on 2020/1/14.
 */
public class InfluxProcess {

    private KMXServerConfig mConfig;
    private KMXClient mClient;

    public InfluxProcess(KMXServerConfig config) {
        mConfig = config;
        mClient = new KMXClient(mConfig.getKMXConfig());
    }


    public void writerInflux(ConsumerRecord<String, String> record){
        String val = record.value();
//        List<Topic1Data> list = (List<Topic1Data>) JSONUtil.JsonToObj(val,List.class);
        List<Topic1Data> list = JSONArray.parseArray(val,Topic1Data.class);
        if (list != null) {
            for (Topic1Data topic1Data : list){
                KMXRecordBuilder builder = null;
                switch (mConfig.getPlatform_type()) {
                    case "tsdb":
                        builder = TSDBRecordBuilder.getBuilder((mConfig.getKMXConfig()));
                        break;
                    case "tsdw":
                        builder = TSDWRecordBuilder.getBuilder((mConfig.getKMXConfig()));
                        break;
                }
                if (builder != null) {
                    builder.setTstable(mConfig.getTable_name());
                    Set<String> tagKeys = topic1Data.getTags().keySet();
                    for (String tagKey : tagKeys) {
                        String tagValue = topic1Data.getTags().get(tagKey);
                        builder.setTag(tagKey, tagValue);
                    }

                    for (String id : topic1Data.getFields().keySet()) {
                        Double value = topic1Data.getFields().get(id);
                        builder.setField(id, value);
                    }
                    builder.setTimestamp(topic1Data.getTime());
                    try {
                        mClient.send(builder.getRecord());
                    } catch (SendException e) {
                        e.printStackTrace();
                    } catch (NullRecordException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}
